package nl.justobjects.pushlet.pushmanager;
import nl.justobjects.pushlet.core.Dispatcher;
import nl.justobjects.pushlet.core.Event;
import nl.justobjects.pushlet.core.Protocol;
import nl.justobjects.pushlet.util.Log;
import nl.justobjects.pushlet.util.Sys;
import nl.justobjects.pushlet.wrapper.JMSService;
import org.springframework.context.ApplicationContext;
import org.springframework.web.context.support.WebApplicationContextUtils;

import java.io.Serializable;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * IDEA 13
 * Created by xnkl on 14-3-10.
 */
public final class PushManager {
	
	public  final long SCAN_TIME_MINS = 5;
	private JMSService jmsService;
	private Map<String, PushThread> eventList = new HashMap<String, PushThread>();
    private final PushThread[] PUSTHREADS = new PushThread[0];
    private static String localMac;
    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
    private ExecutorService exec = Executors.newCachedThreadPool();

	private static class SingletonHolder {
        public final static PushManager instance = new PushManager();    
    } 

    private PushManager() {
    	
    }

    public static PushManager getInstance() {
        return SingletonHolder.instance;
    }
    
    public void start() {
    	// 获取本机MAC
    	localMac = getLocalMac();
    	
    	// 获取JMS服务，如果null则为单机环境
    	ApplicationContext ac = WebApplicationContextUtils.getWebApplicationContext(ApplicationHelper.getServletContext());
    	this.jmsService = (JMSService) ac.getBean("jmsservice");
    	
    	// 每个一段时间回收过期的线程资源
    	scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable() {
			@Override
			public void run() {
				Log.debug("[PushManager]: scheduled clear.");
				long now = Sys.now();
				PushThread[] pushThreads = getEventList();
				for (int i = 0, len = pushThreads.length; i < len; i++) {
					PushThread pushThread = pushThreads[i];
					if (!pushThread.isAlive() || pushThread.getEndTime() <= now) {
						removeEvent(pushThread.getThreadId());
					}
				}
			}
		}, SCAN_TIME_MINS, SCAN_TIME_MINS, TimeUnit.MINUTES);
    }
    
    public void stop() {
        exec.shutdownNow();
		eventList.clear();
    	scheduledThreadPoolExecutor.shutdownNow();
    }
       
    public synchronized PushThread[] getEventList() {
        return eventList.values().toArray(PUSTHREADS);
    }

    /**
     * 创建事件源线程
     * @param threadId 线程ID
     * @param subject 事件主题
     * @param label 事件标签
     * @return 返回一个未启动的线程，使用PushManager.getInstance().startEvent()启动；可以通过PushThread的
     * setPushData，setWhenFinish，setDataCreator，setThreadLife, setDelay等方法设置事件源参数。
     */
    public PushThread createUnstartEvent(String threadId, String subject, String label) {
    	// TODO 合并事件源线程
        PushThread newPushThread = new PushThread(threadId, subject, label).setJMSService(jmsService);
        replaceOldPushThread(threadId, newPushThread);
        return newPushThread;
    }

    /**
     * 创建并启动事件源线程
     * @param threadId 线程ID
     * @param subject 事件主题
     * @param label 事件标签
     * @param data 推送数据
     * @param whenFinish 线程结束时回调
     * @param params 回调函数参数
     * @param interval 推送间隔
     * @param runTime 线程存活时间
     * @return
     */
    public PushThread createEvent(String threadId, String subject, String label,
    		Map<String, ? extends Serializable> data, WhenFinish whenFinish, Map<String, Object> params,
    		long interval, long runTime) {
        Log.debug("new push source thread...");
        PushThread newPushThread = new PushThread(threadId, subject, label).setPushData(data).
                setWhenFinish(whenFinish, params).setThreadLife(interval, runTime).setJMSService(jmsService);
        Log.debug("replace old push source thread...");
        replaceOldPushThread(threadId, newPushThread);
        Log.debug("start push source thread...");
        startEvent(newPushThread);
        return newPushThread;
    }

    /**
     * 创建并启动事件源线程
     * @param threadId 线程ID
     * @param subject 事件主题
     * @param label 事件标签
     * @param dataCreator 推送数据创建函数
     * @param whenFinish 线程结束时回调
     * @param params 推送数据创建函数和线程结束时回调函数参数
     * @param interval 推送时间间隔
     * @param runTime 线程存活时间
     * @return
     */
    public PushThread createEvent(String threadId, String subject, String label, DataCreator dataCreator,
    		WhenFinish whenFinish, Map<String, Object> params, long interval, long runTime) {
        Log.debug("new push source thread...");
        PushThread newPushThread = new PushThread(threadId, subject, label).setDataCreator(dataCreator, params)
                .setWhenFinish(whenFinish, params).setThreadLife(interval, runTime).setJMSService(jmsService);
        Log.debug("replace old push source thread...");
        replaceOldPushThread(threadId, newPushThread);
        Log.debug("start push source thread...");
        startEvent(newPushThread);
        return newPushThread;
    }

    private void replaceOldPushThread(String threadId, PushThread newPushThread) {
        PushThread oldPushThread;
        synchronized (this) {
            oldPushThread = eventList.get(threadId);
            if (oldPushThread != null) {
                // 如果线程已存在，则删除，用新的替换
                eventList.remove(threadId);
                eventList.put(threadId, newPushThread);
            } else {
                eventList.put(threadId, newPushThread);
            }
        }
        if (oldPushThread != null)
            if (oldPushThread.isAlive())
                oldPushThread.stop();
        else
            // 删除集群其他服务器上相同threadId的事件源线程，保证整个系统中只有一个同一事件源
            removeThreadInOtherNodes(threadId);
    }

    /**
     * 启动事件源
     * @param pushThread
     */
    public void startEvent(PushThread pushThread) {
        exec.execute(pushThread);
    }

    /**
     * 删除事件源
     */
    public void removeEvent(String threadId) {
        Log.debug("[PushManager]: Remove PushThread " + threadId);
        synchronized (this) {
            PushThread sm = eventList.get(threadId);
            if (sm == null)
                return;
            if (sm.isAlive())
                sm.stop();
            eventList.remove(threadId);
        }
    }
    
    /**
     * 暂停事件源
     */
    public void pauseEvent(String threadId) {
    	Log.debug("[PushManager]: Pause PushThread " + threadId);
        PushThread sm;
        synchronized (this) {
            sm = eventList.get(threadId);
        }
        if (sm != null && sm.isActive() && sm.isAlive()) {
            sm.passivate();
        }
	}
    
    /**
     * 重启事件源
     */
    public void restartEvent(String threadId) {
    	Log.debug("[PushManager]: Restart PushThread " + threadId);
        PushThread sm;
        synchronized (this) {
            sm = eventList.get(threadId);
        }
        if (sm != null && !sm.isActive() && sm.isAlive()) {
            sm.activate();
        }
    }
    
    public synchronized PushThread getPushThread(String threadId) {
    	return eventList.get(threadId);
    }

    /**
     * 多播消息
     * @param eventSubject
     * @param eventlabel
     * @param map
     */
    public void multicastEvent(String eventSubject, String eventlabel, Map<String, Serializable> map) {
    	Event event = Event.createDataEvent(eventSubject);
        if (eventlabel != null)
            event.setField(Protocol.P_LABEL, eventlabel);
        if (map != null)
        	event.setFields(map);
        Dispatcher.getInstance().multicast(event);
	}
    
    public void multicastEvent(Event event) {
    	Dispatcher.getInstance().multicast(event);
    }

    /**
     * 单播消息
     * @param eventSubject
     * @param eventlabel
     * @param map
     * @param pid
     */
    public void unicastEvent(String eventSubject, String eventlabel, Map<String, Serializable> map, String pid) {
    	Event event = Event.createDataEvent(eventSubject);
    	if (eventlabel != null)
            event.setField(Protocol.P_LABEL, eventlabel);
    	if (map != null)
    		event.setFields(map);
    	Dispatcher.getInstance().unicast(event, pid);
	}
    
    public void unicastEvent(Event event, String pid) {
    	Dispatcher.getInstance().unicast(event, pid);
    }
    
    /**
     * 广播消息，不需要主题、标签
     * @param map
     */
    public void broadcastEvent(Map<String, Serializable> map) {
    	Event event = new Event();
    	event.setFields(map);
    	Dispatcher.getInstance().broadcast(event);
    }
    
    public static class PushManageCommand implements Serializable {
    	public static final String PUSH_MANAGE_CMD_RMTHREAD = "rm_thread";
    	public final String LOCAL_MAC = localMac;
    	private String commandline;
    	private Map<String, String> params = new HashMap<String, String>();
		public String getCommandline() {
			return commandline;
		}
		public PushManageCommand(String commandline) {
			this.commandline = commandline;
		}
		public void setCommandParams(String parameterName, String parameterValue) {
			if (parameterValue != null)
				params.put(parameterName, parameterValue);
		}
		public Map<String, String> getCommandParams() {
			return params; 
		}
    }
    
    private static String getLocalMac() {
		byte[] mac = new byte[0];
		try {
			mac = NetworkInterface.getByInetAddress(InetAddress.getLocalHost()).getHardwareAddress();
		} catch (SocketException e) {
			e.printStackTrace();
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
		StringBuilder sb = new StringBuilder("");
		for(int i = 0, len = mac.length; i < len; i++) {
			if(i != 0) {
				sb.append("-");
			}
			//字节转换为整数
			int temp = mac[i] & 0xff;
			String str = Integer.toHexString(temp);
			if(str.length() == 1) {
				sb.append("0").append(str);
			}else {
				sb.append(str);
			}
		}
		return sb.toString().toUpperCase();
	}

    public void doCommand(PushManageCommand command) {
    	String cmdline = command.getCommandline();
        Log.debug("[PushManager]: doCommand, commandline source mac - " + command.LOCAL_MAC);
    	Map<String, String> params = command.getCommandParams();
    	// 本机命令不执行
    	if (localMac.equals(command.LOCAL_MAC))
    		return;
    	if (PushManageCommand.PUSH_MANAGE_CMD_RMTHREAD.equals(cmdline)) {
    		String threadId = params.get("threadId");
	    	removeEvent(threadId);
    	}
    }
    
    /**
     * 删除集群节点上指定threadId的线程
     * @param threadId
     */
    private void removeThreadInOtherNodes(String threadId) {
    	if (jmsService != null) {
    		PushManageCommand command = new PushManageCommand(PushManageCommand.PUSH_MANAGE_CMD_RMTHREAD);
    		command.setCommandParams("threadId", threadId);
    		jmsService.broadcastPushManageCommand(command);
    	}
    }
}
